package com.atguigu.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**
 * 制定时间进项消费
 */
public class CustomConsumerSeekForTime {

    public static void main(String[] args) {
// 0. 配置
        Properties properties = new Properties();

        // 连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.12:9092");
        // 反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 设置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");
        // 关闭自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);


        // 1. 创建kafka消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 2. 订阅主题
        kafkaConsumer.subscribe(Collections.singleton("first"));

        // 指定位置进行消费
        Set<TopicPartition> assignment = kafkaConsumer.assignment(); // 获取对应的分区信息
        // 保证分区分配方案已经制定完毕
        while(assignment.size()==0){
            kafkaConsumer.poll(Duration.ofSeconds(1));
            assignment = kafkaConsumer.assignment();
        }

        // 希望把时间转换成offset
        HashMap<TopicPartition, Long> topicPartitionLongHashMap = new HashMap<>();
        // 封装对应集合
        for (TopicPartition topicPartition : assignment) {
            // 一天前的offset位置
            topicPartitionLongHashMap.put(topicPartition, System.currentTimeMillis() - 24 * 3600 * 100);
        }
        Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = kafkaConsumer.offsetsForTimes(topicPartitionLongHashMap);

        // 指定消费的offset, 通过时间转换获取到offset
        for (TopicPartition topicPartition : assignment) {
            OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);
            kafkaConsumer.seek(topicPartition, offsetAndTimestamp.offset());
        }

        // 3. 消费消息
        while(true){

            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
            kafkaConsumer.commitAsync();

        }


    }
}
